package com.example.demo.study.sceneQuestion.Q1;

import com.example.demo.study.sceneQuestion.Q1.data.PoisonPill;
import com.example.demo.study.sceneQuestion.Q1.data.ProcessedData;
import com.example.demo.study.sceneQuestion.Q1.data.RawData;
import com.example.demo.study.sceneQuestion.Q1.parser.DataParser;
import com.example.demo.study.sceneQuestion.Q1.reader.DataReader;
import com.example.demo.study.sceneQuestion.Q1.reader.FileDataReader;
import com.example.demo.study.sceneQuestion.Q1.writer.DataWriter;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

// 数据处理主类
public class DataProcessor {
    // 定义线程池
    private final ExecutorService readerExecutor;
    private final ExecutorService parserExecutor;
    private final ExecutorService writerExecutor;

    // 阻塞队列用于传递数据
    private final BlockingQueue<RawData> rawDataQueue;
    private final BlockingQueue<ProcessedData> processedDataQueue;

    // 用于标记是否还有数据需要处理
    private final AtomicBoolean isReadingComplete = new AtomicBoolean(false);

    // 构造函数，初始化线程池和队列
    public DataProcessor(int readerThreads, int parserThreads, int writerThreads,
                         int rawQueueSize, int processedQueueSize) {
        // 初始化线程池
        readerExecutor = Executors.newFixedThreadPool(readerThreads);
        parserExecutor = Executors.newFixedThreadPool(parserThreads);
        writerExecutor = Executors.newFixedThreadPool(writerThreads);

        // 初始化阻塞队列
        rawDataQueue = new ArrayBlockingQueue<>(rawQueueSize);
        processedDataQueue = new ArrayBlockingQueue<>(processedQueueSize);

        // 启动解析器和写入器的工作线程
        startParsers(parserThreads);
        startWriters(writerThreads);
    }

    // 添加数据读取任务
    public void addReadingTask(DataReader reader) {
        readerExecutor.submit(() -> {
            try {
                RawData data;
                // 读取数据直到没有更多数据
                while ((data = reader.read()) != null) {
                    // 将原始数据放入队列，会阻塞直到有空间
                    rawDataQueue.put(data);
                }
            } catch (Exception e) {
                System.err.println("读取数据时发生错误: " + e.getMessage());
                e.printStackTrace();
            } finally {
                // 当一个读取器完成，检查是否所有读取器都已完成
                if (readerExecutor.isTerminated()) {
                    isReadingComplete.set(true);
                }
            }
        });
    }

    // 启动解析器工作线程
    private void startParsers(int count) {
        for (int i = 0; i < count; i++) {
            parserExecutor.submit(() -> {
                try {
                    while (true) {
                        // 从原始数据队列中获取数据，如果队列空且读取已完成则退出
                        RawData rawData = rawDataQueue.poll(1, TimeUnit.SECONDS);
                        if (rawData == null) {
                            if (isReadingComplete.get()) {
                                break;
                            }
                            continue;
                        }

                        // 解析数据
                        DataParser parser = new DataParser();
                        ProcessedData processedData = parser.parse(rawData);

                        // 将解析后的数据放入队列
                        processedDataQueue.put(processedData);
                    }
                } catch (Exception e) {
                    System.err.println("解析数据时发生错误: " + e.getMessage());
                    e.printStackTrace();
                } finally {
                    // 检查是否所有解析器都已完成
                    if (parserExecutor.isTerminated()) {
                        // 标记所有数据已解析完成
                        try {
                            processedDataQueue.put(new PoisonPill());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
        }
    }

    // 启动写入器工作线程
    private void startWriters(int count) {
        for (int i = 0; i < count; i++) {
            writerExecutor.submit(() -> {
                try {
                    while (true) {
                        // 从解析后的数据队列中获取数据
                        ProcessedData data = processedDataQueue.take();

                        // 如果是终止标记，则退出
                        if (data instanceof PoisonPill) {
                            // 将终止标记放回队列，让其他写入线程也能收到
                            processedDataQueue.put(data);
                            break;
                        }

                        // 存储数据
                        DataWriter writer = new DataWriter();
                        writer.write(data);
                    }
                } catch (Exception e) {
                    System.err.println("存储数据时发生错误: " + e.getMessage());
                    e.printStackTrace();
                }
            });
        }
    }

    // 等待所有任务完成并关闭资源
    public void waitForCompletion() {
        try {
            // 等待所有读取任务完成
            readerExecutor.shutdown();
            readerExecutor.awaitTermination(1, TimeUnit.HOURS);

            // 等待所有解析任务完成
            parserExecutor.shutdown();
            parserExecutor.awaitTermination(1, TimeUnit.HOURS);

            // 等待所有写入任务完成
            writerExecutor.shutdown();
            writerExecutor.awaitTermination(1, TimeUnit.HOURS);
        } catch (InterruptedException e) {
            System.err.println("处理被中断: " + e.getMessage());
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) {
        // 创建数据处理器实例
        DataProcessor processor = new DataProcessor(
                2,  // 读取线程数
                4,  // 解析线程数
                2,  // 写入线程数
                100, // 原始数据队列大小
                100  // 处理后数据队列大小
        );

        // 添加数据读取任务
        processor.addReadingTask(new FileDataReader("data1.txt"));
        processor.addReadingTask(new FileDataReader("data2.txt"));

        // 等待所有任务完成
        processor.waitForCompletion();
        System.out.println("所有数据处理完成");
    }
}
